package com.yanggu.mapreduce.customizeoutputformat;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;

/**
 * @version V1.0
 * @author: YangGu
 * @date: 2020/2/7 22:03
 * @description:
 */
public class FilterRecordWriter extends RecordWriter<Text, NullWritable> {

    FSDataOutputStream atGuiOutPutStream;
    FSDataOutputStream otherOutPutStream;

    public FilterRecordWriter(TaskAttemptContext job) {
        FileSystem fileSystem = null;
        try {
            fileSystem = FileSystem.get(job.getConfiguration());
            atGuiOutPutStream = fileSystem.create(new Path("output/customizeoutputformat/atgui.log"));
            otherOutPutStream = fileSystem.create(new Path("output/customizeoutputformat/other.log"));
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (fileSystem != null) {
                try {
                    fileSystem.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    @Override
    public void write(Text key, NullWritable value) throws IOException, InterruptedException {
        String line = key.toString();
        if (line.contains("atgui")) {
            atGuiOutPutStream.write(line.getBytes());
        } else {
            otherOutPutStream.write(line.getBytes());
        }
    }

    @Override
    public void close(TaskAttemptContext context) throws IOException, InterruptedException {
        IOUtils.closeStream(atGuiOutPutStream);
        IOUtils.closeStream(otherOutPutStream);
    }
}
